/************************************************************************
Copyright 2003, 2004, 2005, 2006, 2007, 2008 Grid Technology Research Center,
National Institute of Advanced Industrial Science and Technology.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
************************************************************************/

package jp.aist.gtrc.plus.reserve;

import java.util.Calendar;
import java.util.Collection;
import java.util.LinkedList;

import static jp.aist.gtrc.plus.reserve.ReserveStatus.State.*;
import jp.aist.gtrc.plus.reserve.ReserveStatus.State;
import jp.aist.gtrc.plus.scheduler.ScheduleStarter;
import jp.aist.gtrc.plus.scheduler.SchedulerOptions;
import jp.aist.gtrc.plus.scheduler.node.NodeAllocator;
import jp.aist.gtrc.plus.scheduler.node.NodeInfo;
import jp.aist.gtrc.plus.scheduler.node.NodeManager;
import jp.aist.gtrc.plus.scheduler.specific.sge.qbase.SGEQbaseMainServer;
import jp.aist.gtrc.plus.scheduler.status.QInstanceStatus;
import jp.aist.gtrc.plus.scheduler.status.QueueStatus;
import jp.aist.gtrc.plus.scheduler.status.StatusManager;
import jp.aist.gtrc.plus.scheduler.sys.SchedulerException;

public abstract class QbaseReserveManager extends ReserveManager {

	public static final String POSTFIX_PREQ = "pre";
	public static final String POSTFIX_POSTQ = "post";

	protected enum Phase {
		ActivatePre(0),
		DelPreActMain(1),
		ActivateMain(2),
		DelMainActPost(3),
		DeletePost(4),
		DeleteMain(5),
		DeleteAll(6),
		DeleteSuper(7);

		private final int phase;
		private Phase(int phase) {
			this.phase = phase;
		}
		public final int getPhase() {
			return phase;
		}
	}

	public QbaseReserveManager(NodeManager nodeMgr, ScheduleStarter starter,
			ReserveRecorder recorder, NodeAllocator nodeAllocator,
			StatusManager statusMgr, SchedulerOptions options) throws ReserveException {

		super(nodeMgr, starter, recorder, nodeAllocator, options, statusMgr);

	}

	protected abstract void suspendQInstances(Collection<QInstanceStatus> qinsList);
	protected abstract void resumeQInstances(Collection<QInstanceStatus> qinsList);

	protected void scheduleWhenIdle() throws ReserveException {

		Calendar now = Calendar.getInstance();
		ReserveTable table = getReserveTable();

		/*
		 * Schedule Reserve queue
		 */
		// NOTE: make a clone list because some element will be removed.
		Collection<ReserveInfo> allRsvs = new LinkedList<ReserveInfo>(table.getAllById());
		for (ReserveInfo info : allRsvs) {
			State state = info.getReserveState();
			if (info.isExpired(now) || (state == Canceled)) {
				stopReservePeriodWithoutStatusUpdate(info);
			}
		}
		Collection<ReserveInfo> expires = getExpiredReserve(now);
		for (ReserveInfo info : table.getAllById()) {
			if (info.isOverlap(now, now)) {
				startReservePeriod(info, expires);
			}
		}

		/*
		 * Schedule normal queue
		 */
		LinkedList<QInstanceStatus> toBeResumed
			= new LinkedList<QInstanceStatus>();
		LinkedList<QInstanceStatus> toBeSuspended
			= new LinkedList<QInstanceStatus>();

		StatusManager statusMgr = getStatusManager();
		for (QueueStatus queue : statusMgr.getQueues()) {
			if (isReserveQueue(queue)) {
				continue;
			}

			NodeManager nodeMgr = getNodeManager();
			// resume/suspend normal queue if needed
			for (QInstanceStatus qins : queue.getAllQInstances()) {
				String nodeName =  qins.getHostname();
				NodeInfo node = nodeMgr.getNode(nodeName);
				if (node == null) {
					assert(false);
					continue;
				}
				if (node.isFree(now, now)) {
					if (qins.isSuspended()) {
						toBeResumed.add(qins);
					}
				} else {
					if (qins.isReadyToUse()) {
						toBeSuspended.add(qins);
					}
				}
			}
		}

		if (toBeResumed.isEmpty() == false) {
			resumeQInstances(toBeResumed);
		}
		if (toBeSuspended.isEmpty() == false) {
			suspendQInstances(toBeSuspended);
		}

	}

	public synchronized void schedule() throws ReserveException {

		if (getCurrentOperation().getOperationType() == ReserveOperationType.idle) {
			scheduleWhenIdle();
		} else {
			scheduleWaitingCommitAbort();
		}

	}

	protected String getMainQName(ReserveInfo info) {
		return info.getReserveId().toString();
	}

	protected boolean isReserveQueue(QueueStatus queue) {
		return queue.getName().startsWith(ReserveId.RSVID_PREFIX);
	}

	protected String getPreQName(ReserveInfo info) {
		return getMainQName(info) + POSTFIX_PREQ;
	}

	protected String getPostQName(ReserveInfo info) {
		return getMainQName(info) + POSTFIX_POSTQ;
	}

	protected QueueStatus getQueue(String qName) {
		StatusManager statusMgr = getStatusManager();
		return statusMgr.getQueue(qName);
	}

	protected boolean isSuspended(QueueStatus q) {

		for (QInstanceStatus qi : q.getAllQInstances()) {
			if (qi.isSuspended())
				return true;
		}
		return false;

	}

	protected abstract boolean hasJobToRun(QueueStatus q);

	protected boolean isPreviousPostQRemain(ReserveInfo info,
			Collection<ReserveInfo> expires) {

		ReserveNodeSet tgtNodes = info.getReserveNodeSet();
		String[] tgtNames = tgtNodes.getNames();

		for (ReserveInfo rsv : expires) {
			QueueStatus queue = getQueue(getPostQName(rsv));
			if (queue == null) {
				continue;
			}
			for (QInstanceStatus qins : queue.getAllQInstances()) {
				String qInsHostName = qins.getHostname();
				for (String hostname : tgtNames) {
					if (hostname.equals(qInsHostName)) {
						return true;
					}
				}
			}
		}
		return false;

	}

	protected Phase getStartPhase(ReserveInfo info, Collection<ReserveInfo> expires) {

		switch (info.getReserveState()) {
		case Confirmed:
		case Running:
			// Start this reserve if needed
			break;
		case Done:
		case Canceled:
		case Destroying:
		case Error:
			// DO NOT start this reserve
			return null;
		default:
			assert(false);
			return null;
		}

		if (isPreviousPostQRemain(info, expires)) {
			// We cannot start reserve of 'info'
			// until previous postQ will be gone...
			return null;
		}

		QueueStatus mainQ = getQueue(getMainQName(info));

		if (info.getOptions().isMakePrePostQueue()) {
			QueueStatus preQ = getQueue(getPreQName(info));
			if (preQ != null) {
				if (hasJobToRun(preQ)) {
					if (isSuspended(preQ)) {
						// pre queue is suspended, activate now top run pre job!
						return Phase.ActivatePre;
					} else {
						// pre job is running (or wait to run), nothing to do
						return null;
					}
				} else {
					if (hasJobToRun(mainQ)) {
						// pre job is finished, delete preQ and resume mainQ now!
						return Phase.DelPreActMain;
					} else {
						// pre and main job is empty. Wait job submition to pre/mainQ.
						return null;
					}
				}
			}
		}

		if ((mainQ != null) && isSuspended(mainQ)) {
			return Phase.ActivateMain;
		} else {
			return null;
		}

	}

	protected abstract void resumeQueue(ReserveInfo info, String qname) throws ReserveException;
	protected abstract void deleteQueue(ReserveInfo info, String qname) throws ReserveException;

	protected void startReservePeriod(ReserveInfo info, Phase phase) throws ReserveException {

		String mainQName = getMainQName(info);
		String preQName = getPreQName(info);

		switch (phase) {
		case ActivatePre:
			resumeQueue(info, preQName);
			break;
		case DelPreActMain:
			deleteQueue(info, preQName);
			resumeQueue(info, mainQName);
			break;
		case ActivateMain:
			resumeQueue(info, mainQName);
			break;
		default:
			assert(false);
			break;
		}

	}

	protected void finishReservePeriod(ReserveInfo info, Phase phase) throws ReserveException {

		String mainQName = getMainQName(info);
		String preQName = getPreQName(info);
		String postQName = getPostQName(info);

		switch (phase) {
		case DelMainActPost:
			deleteQueue(info, mainQName);
			resumeQueue(info, postQName);
			break;
		case DeletePost:
			deleteQueue(info, postQName);
			break;
		case DeleteMain:
			deleteQueue(info, mainQName);
			break;
		case DeleteAll:
			deleteQueue(info, preQName);
			deleteQueue(info, mainQName);
			deleteQueue(info, postQName);
			break;
		default:
			assert(false);
			break;
		}

	}

	protected void startReservePeriod(ReserveInfo info, Collection<ReserveInfo> expires)
		throws ReserveException {

		Phase phase = getStartPhase(info, expires);
		if (phase == null)
			return;

		switch (phase) {
		case ActivatePre:
		case DelPreActMain:
		case ActivateMain:
			getReserveNodeNormalizer().reallocateJustBeforeStart(info);
			startReservePeriod(info, phase);
			break;
		default:
			assert(false);
			break;
		}

	}

	protected Phase getStopPhase(ReserveInfo info) {

		if (info.getOptions().isMakePrePostQueue()) {
			QueueStatus preQ = getQueue(getPreQName(info));
			if (preQ != null) {
				if (isSuspended(preQ)) {
					// pre has not started, delete all queue now
					return Phase.DeleteAll;
				} else if (hasJobToRun(preQ)) {
					// pre job is running now, wait it finish.
					return null;
				} else {
					// pre has already finished, handle main or post queue
				}
			}

			QueueStatus postQ = getQueue(getPostQName(info));
			if (postQ != null) {
				if (hasJobToRun(postQ)) {
					if (isSuspended(postQ)) {
						// post queue is suspended, disable mainQ and resume postQ
						return Phase.DelMainActPost;
					} else {
						// post job is running (or wait to run), nothing to do
						return null;
					}
				} else {
					// post job is finished or post job doesn't exist, delete all!
					return Phase.DeleteAll;
				}
			} else {
				// post queue is already deleted, mainQ is already deleted
				// call super.remove() to remove from DB
				return Phase.DeleteSuper;
			}
		} else {
			QueueStatus mainQ = getQueue(getMainQName(info));
			return (mainQ != null) ? Phase.DeleteMain : Phase.DeleteSuper;
		}

	}

	protected void stopReservePeriod(ReserveInfo info) throws ReserveException {

		try {
			StatusManager statusMgr = getStatusManager();
			statusMgr.updateStatus(SGEQbaseMainServer.getInstance());
		} catch (SchedulerException e) {
			throw new ReserveException(e.getMessage());
		}

		stopReservePeriodWithoutStatusUpdate(info);

	}

	protected void stopReservePeriodWithoutStatusUpdate(ReserveInfo info) throws ReserveException {

		Phase phase = getStopPhase(info);
		if (phase == null)
			return;

		switch (phase) {
		case DelMainActPost:
			finishReservePeriod(info, phase);
			break;
		case DeletePost:
		case DeleteMain:
		case DeleteAll:
			super.removeReserveInfo(info, false);
			finishReservePeriod(info, phase);
			break;
		case DeleteSuper:
			// pre/main/postQ is already deleted, delete from super class
			super.removeReserveInfo(info, false);
			break;
		default:
			assert(false);
			break;
		}

	}

	protected void removeReserveInfo(ReserveInfo info, boolean forceRemove)  throws ReserveException {

		if (forceRemove) {
			Phase phase = info.getOptions().isMakePrePostQueue()
				? Phase.DeleteAll : Phase.DeleteMain;
			finishReservePeriod(info, phase);
			super.removeReserveInfo(info, forceRemove);
		} else {
			stopReservePeriod(info);
		}

	}

}
